package com.cayden.componets.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.cayden.componets.canal.handle.CanalContext;
import com.cayden.componets.canal.handle.CanalFirstHandleContextInfo;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;

import java.util.List;

/**
 * 基类
 * 对于canal的抽象
 * @author cayden
 * @since 2021/10/19
 */
public class AbstractCanalClient extends BaseCanalClient {

    @Setter
    @Getter
    protected ThreadPoolTaskExecutor executor;

    @Autowired
    protected CanalFirstHandleContextInfo canalFirstHandleContextInfo;


    public AbstractCanalClient(String destination) {
        this(destination, null);
    }

    public AbstractCanalClient(String destination, CanalConnector connector) {
        this.destination = destination;
        this.connector = connector;
    }

    protected void start() {
        Assert.notNull(connector, "connector is null");
        thread = new Thread(new Runnable() {

            @Override
            public void run() {
                process();
            }
        });

        thread.setUncaughtExceptionHandler(handler);
        running = true;
        thread.start();
    }

    protected void stop() {
        if (!running) {
            return;
        }
        running = false;
        if (thread != null) {
            try {
                thread.join();
            } catch (InterruptedException e) {
                // ignore
            }
        }

        MDC.remove("destination");
    }

    protected void process() {
        int batchSize = 5 * 1024;
        logger.info("初始化init --------------------------------");
        while (running) {
            try {
                MDC.put("destination", destination);
                connector.connect();
                connector.subscribe();
                while (running) {
                    //logger.info("连接成功，这里已经在执行while循环获取");
                    // 获取指定数量的数据
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    //logger.info("Message.size:"+size);
                    if (batchId != -1 && size != 0) {
                        MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, executor);
                        List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(datas, message.getId());
                        for (FlatMessage flatMessage : flatMessages) {
                            executor.submit(new Runnable() {
                                @Override
                                public void run() {
                                    CanalContext context = new CanalContext();
                                    context.setFlatMessage(flatMessage);
                                    canalFirstHandleContextInfo.doChangeHandleInfo(context);
                                }
                            });
                        }
                    } else {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                        }
                    }

                    if (batchId != -1) {
                        // 提交确认
                        //logger.info("提交确认:");
                        connector.ack(batchId);
                        // connector.rollback(batchId); // 处理失败, 回滚数据
                    }
                }
            } catch (Exception e) {
                logger.error("Canal connect error!");
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e1) {
                    // ignore
                }
            } finally {
                connector.disconnect();
                MDC.remove("destination");
            }
        }
    }

}
